package com.zk.config.api.client;

import java.io.IOException;
import org.apache.log4j.Logger;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import com.zk.config.api.constants.Constants;
import com.zk.config.api.service.ZkCache;
import lombok.Data;

@Data
public class ConfigClient {
	
	private static Logger logger = Logger.getLogger(ConfigClient.class);
	
	private ZooKeeper zk;
	
	private Watcher watcher;
	
	private boolean isClose = true;
	
	private Constants constants = new Constants();
	
	private ZkCache zkCache;
	
	private TimeThread timeThread;
	
	public ConfigClient(String zkHost, int connectTimeout) {
		this(zkHost, connectTimeout, 0, false);
	}
	
	public ConfigClient(String zkHost, int connectTimeout, long refreshDataTime, boolean watch) {
		constants.setZkHost(zkHost);
		constants.setConnectTimeout(connectTimeout);
		constants.setRefreshDataTime(refreshDataTime);
		constants.setWatch(watch);
		init();
	}
	
	public ZooKeeper getZk() {
		if(zk == null && !isClose) {
			try {
				setZk(new ZooKeeper(constants.getZkHost(), constants.getConnectTimeout(), watcher));
				isClose = false;
			} catch (IOException e) {
				e.printStackTrace();
				logger.error(e.getMessage());
			}
		}
		return zk;
	}

	public void setZk(ZooKeeper zk) {
		this.zk = zk;
	}
	
	public void close(){
		try {
			if (zk != null) {
				zk.close();
			}
			isClose = true;
			if (timeThread != null) {
				timeThread.close();
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
			logger.error(e.getMessage());
		}
	}
	
	public void init() {
		String newZkHost = constants.getZkHost().replaceAll("/+", "/").replaceAll("/$", "");
		
		zkCache = new ZkCache(zk, constants.isWatch());
		if (constants.isWatch()) {
			this.watcher = new Watcher() {
				@Override
				public void process(WatchedEvent event) {					
					String path = event.getPath();
					if (event.getType() != EventType.None) {
						if(event.getType() == EventType.NodeDeleted) {
							zkCache.reloadIfNodeDelete(path);
						} else if (event.getType() == EventType.NodeCreated || event.getType() == EventType.NodeDataChanged) {
							try {
								zk.exists(path, true); // 回调继续监听监听变更的路径节点
							} catch (KeeperException | InterruptedException e) {
								e.printStackTrace();
							}
							zkCache.reloadIfNodeCreateOrChange(path);
						} else if(event.getType() == EventType.NodeChildrenChanged) {
							try {
								if (zk.exists(path, false) != null) {
									zk.getChildren(path, true); //回调继续监听父节点变化
								}
							} catch (KeeperException | InterruptedException e) {
								e.printStackTrace();
							}
							zkCache.read(path,false);
						}
					}
				}
			};
		}
		try {
			zk = new ZooKeeper(newZkHost, constants.getConnectTimeout(), watcher);
			isClose = false;
		} catch (IOException e) {
			e.printStackTrace();
			logger.error(e.getMessage());
			return;
		}
		zkCache.setZk(zk);
		zkCache.init();
		
		if(constants.getRefreshDataTime() > 0) {
			timeThread = new TimeThread(constants.getRefreshDataTime());
			timeThread.start();
		}
	}
	
	class TimeThread extends Thread {
		private long time;
		private boolean runflag = true;
		public TimeThread(long time) {
			this.time = time;
		}
		public void run() {
			while(runflag) {
				try {
					Thread.sleep(time);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				zkCache.refresh();
			}
		}
		public void close(){
			runflag = false;
			this.interrupt();
		}
	}
}
